package com.shihb.akka.sparkmasrterwork.worker

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.shihb.akka.sparkmasrterwork.common.{HeartBeat, RegisterWorkerInfo, SendHeartBeat}
import com.typesafe.config.{Config, ConfigFactory}

/**
 * Description:
 *
 * @author shihb
 * @date 2020/10/3 11:36
 */
class SparkWorker(masterHost:String,masterPort:Int,masterName:String) extends  Actor{
  var sparkProxy:ActorSelection =_
  val id=java.util.UUID.randomUUID().toString
  override def receive: Receive = {
    case "start" =>{//开始启动
      println("worker 启动了")
      sparkProxy ! RegisterWorkerInfo(id,16,16*1024)
    }
    case RegisterWorkerInfo => {//收到服务端注册成功信息
      println(id+"注册成功")
      // 注册成功就注册定时器发送心跳
      // 时间单位的包
      import scala.concurrent.duration._
      import context.dispatcher
      //说明 0 millis不延时,3 second每隔3秒,self 发给给自己，发送的消息
      context.system.scheduler.schedule(0 millis,3 second,self,SendHeartBeat)
    }
    case SendHeartBeat=>{ //收到发送心跳信息
      println("worker"+id+"发送了心跳")
      sparkProxy ! HeartBeat(id)
    }
  }

  override def preStart(): Unit = {
    val path=s"akka.tcp://SparkMaster@$masterHost:$masterPort/user/$masterName"
    this.sparkProxy=context.actorSelection(path)
  }
}
object  SparkWorker {
  def main(args: Array[String]): Unit = {
    if(args.length!=6){
      println("请输入参数workerhost workerport workerName masterhost masterport masterName")
      sys.exit()
    }
    val host=args(0)
    val port=args(1).toInt
    val name=args(2)
    val masterHost=args(3)
    val masterPort=args(4).toInt
    val masterName=args(5)






    //创建ActorSystem
    val config: Config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider="akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname=$host
         |akka.remote.netty.tcp.port=$port
         |""".stripMargin)
    val workerActorSystem: ActorSystem = ActorSystem("SparkWorker", config)
    val actorRef: ActorRef = workerActorSystem.actorOf(Props(new SparkWorker(masterHost, masterPort,masterName)), name)

    actorRef!"start"

  }
}
